#include <sys/eventfd.h>
#include <unistd.h>
#include<fcntl.h>
#include <errno.h>
#include <memory>

#include "EventLoop.h"
#include "Channel.h"
#include "Logger.h"
#include "Poller.h"

// 防止一个线程创建多个EventLoop
__thread EventLoop* t_loopInThisThread = nullptr;

// 定义默认的Poller I/O复用的超时时间10s
const int kPollTimeMs = 10000; 

EventLoop::EventLoop()
: looping_(false)
, quit_(false)
, callingPendingFunctors_(false)
, threadId_(CurrentThread::tid())
, poller_(Poller::newDefaultPoller(this))
, wakeupFd_(EventLoop::createEventfd())
, wakeupChannel_(new Channel(this,wakeupFd_))
{
    LOG_DEBUG("EventLoop created %p in thread %d \n",this,threadId_);
    if(t_loopInThisThread)
    {
        LOG_FATAL("Another EventLoop %p exists in this thread %d \n",t_loopInThisThread,threadId_);
    }
    else
    {
        t_loopInThisThread = this;
    }

    // 设置wakeupfd的事件类型以及发生事件后的回调操作
    wakeupChannel_->setReadCallback(std::bind(&EventLoop::handleRead,this));
    // 每一个EventLoop都将监听wakeupChannel的EPOLLIN读事件
    wakeupChannel_->enableReading();

}
 
EventLoop::~EventLoop()
{
    wakeupChannel_->disableAll();
    wakeupChannel_->remove();
    ::close(wakeupFd_);
    t_loopInThisThread = nullptr;
}

void EventLoop::loop()
{
    looping_=true;
    quit_=false;

    LOG_INFO("EventLoop %p start looping \n",this);

    while(!quit_)
    {
        activeChannels_.clear();
        // 等待事件发生 并将活跃事件对应的Channel收集起来
        pollReturnTime_=poller_->poll(kPollTimeMs,&activeChannels_);
        for(Channel* channel : activeChannels_)
        {
            // poller监听那些channel发生事件了 然后上报给EventLoop 通知channel处理相应的事件
            channel->handleEvent(pollReturnTime_);
        }
        // 执行当前EventLoop事件循环需要处理的回调操作
        doPendingFunctors();
    }
    LOG_INFO("EventLoop %p stop looping \n",this);
    looping_=false;
}

void EventLoop::quit()
{
    quit_=true;
    /*
    此处isInLoopThread()返回false 必须调用wakeup()
    因为EventLoop线程可能正阻塞在poll()中
    如果不唤醒 则会一直阻塞 无法检测到quit_=true 导致无法退出
    wakeup()会强制让poll()返回 使EventLoop 线程有机会检查quit_并退出
    */
    if(!isInLoopThread())
    {
        wakeup();
    }
}

void EventLoop::runInLoop(Functor cb)
{
    // 在当前loop中执行cb
    if(isInLoopThread())
    {
        cb();
    }
    // 在非当前loop中 需要唤醒loop所在线程 执行cb
    else
    {
        queueInLoop(cb);
    }
}

// 把回调函数Functor加入待执行队列 并在满足以下任意之一 唤醒事件循环线程执行这些回调
// 1.当前调用 queueInLoop 的线程不是事件循环线程本身
// 2.事件循环线程正在执行之前的回调
void EventLoop::queueInLoop(Functor cb)
{
    {
        std::unique_lock<std::mutex> lock(mutex_);
        pendingFuntors_.emplace_back(cb);
    }

    // 唤醒相应需要执行上述回调操作的loop线程
    if(!isInLoopThread() || callingPendingFunctors_)
    {
        wakeup();
    }
}


// 创建wakeupfd 用来唤醒subReactor处理新来的channel
int EventLoop::createEventfd()
{
    int evtfd=::eventfd(0,EFD_NONBLOCK | EFD_CLOEXEC);
    if(evtfd<0)
    {
        LOG_FATAL("eventfd error:%d \n",errno);
    }
    return evtfd;
}

void EventLoop::handleRead()
{
    uint64_t one = 1;
    ssize_t n = read(wakeupFd_,&one,sizeof one);
    if(n!=sizeof one)
    {
        LOG_ERROR("EventLoop::handleRead() reads %zd bytes instead of 8",n);

    }
}

// 用来唤醒loop所在的线程 向wakeupFd_写一个数据 wakeupChannel发生读事件 当前loop线程就会被唤醒
void EventLoop::wakeup()
{
    uint64_t one=1;
    ssize_t n=write(wakeupFd_,&one,sizeof one);
    if(n!=sizeof one)
    {
        LOG_ERROR("EventLoop::wakeup() writes %lu bytes instead of 8 \n",n);
    }
}

void EventLoop::updateChannel(Channel* channel)
{
    poller_->updateChannel(channel);
}

void EventLoop::removeChannel(Channel* channel)
{
    poller_->removeChannel(channel);
}

bool EventLoop::hasChannel(Channel* channel)
{
    return poller_->hasChannel(channel);
}

// 执行回调
void EventLoop::doPendingFunctors()
{
    std::vector<Functor> functors;
    callingPendingFunctors_=true;

    {
        std::unique_lock<std::mutex> lock(mutex_);
        functors.swap(pendingFuntors_);
    }

    for(const Functor& functor : functors)
    {
        functor(); // 执行当前loop需要的回调操作
    }

    callingPendingFunctors_=false;

}